#ifndef TRANSFORMERCONTENTPORTSTREAMIN_HPP
#define TRANSFORMERCONTENTPORTSTREAMIN_HPP

#include "logger.hpp"
#include "transContentDefine.h"
#include "transContentErrorCode.h"
#include "transContentDispatcher.hpp"
#include "transContentMemory.hpp"
#include "transContentSetting.hpp"

class TransStreamInMemory : public TransContentITimerEvent{
	private:

		int mCapacity;

		/// insert data index in buffer
		int mInsertIndex;

		/// avaliable data in buffer index
		int mAvaliableIndex;

		/// avaliable streamId start value, if(avaliableindex == insertIndex && [insertIndex] != -1) full
		int mStreamValueStart;

		/// buffer left buffer count
		int mLeftCount;

		/// avaliable data buffer count == capacity is full
		int mAvaliableCount;

		/// when get data buffer size less than the buffer size, each can not get a whole buffer, so this value to set next time start offset
		int mRecordInnerOffset;

		/// the buffer left value bytes to read, default is the record data length
		int mRecordLeftBytes;

		/// the buffer full, but the caller not read, elapsetime, larger than some second, will close this port
		int mDataAvaliableFullElapseTime;

		/// wait data avaliable timeout
		int timeout;

		/// the store data buffer pool
		short* recordBuffer;

		/// receive data callback
		recvSyncCallback receiveCallback;

		/// wait data avaliable condtion value
		pthread_cond_t waitCond;

		/// wait data avalibale mutex
		pthread_mutex_t waitMutex;

		///access buffer pool mutex
		pthread_mutex_t accessMemoryPoolMutex;

		char* avaliableData;

		NodeAddr target;

		int remotePort;

		vroute_t vid;

	public:
		///
		/// the timer interface callback
		void OnTime();

		TransStreamInMemory(): mCapacity(STREAM_BUFFER_SIZE_DEFAULT), mInsertIndex(0), mAvaliableIndex(0), mStreamValueStart(0), mLeftCount(mCapacity), mAvaliableCount(0), mRecordInnerOffset(STREAM_DATA_START_INDEX), mRecordLeftBytes(0), mDataAvaliableFullElapseTime(0), timeout(INT_MAX), recordBuffer(NULL), receiveCallback(NULL), waitCond(PTHREAD_COND_INITIALIZER), waitMutex(PTHREAD_MUTEX_INITIALIZER), accessMemoryPoolMutex(PTHREAD_MUTEX_INITIALIZER), avaliableData(NULL){
			debug_enter();
			recordBuffer = (short*)calloc(mCapacity, sizeof(short));
			if(recordBuffer == NULL){
				logd_fata_error("stream in memory, alloc memory failed: %s", strerror(errno));
				return;
			}
			logd_info("stream in memory, reset the memory to index to -1");
			for(int i=0; i< mCapacity; ++i){
				recordBuffer[i] = -1;
			}
			debug_leave();
		}

		~TransStreamInMemory(){
			debug_enter();
			logd_info("stream in memory, start to release the buffer pool");
			if(recordBuffer != NULL){
				free(recordBuffer);
				recordBuffer = NULL;
			}
			debug_leave();
		}

		void setTarget(const NodeAddr* target){
			this->target = *target;
		}

		void setVid(vroute_t vid){
			this->vid = vid;
		}

		void setRemotePort(int remotePort){
			this->remotePort = remotePort;
		}

		int getCapacity() const {
			return mCapacity;
		}

		///
		/// get left size to store data
		///
		int getAvalibaleBytes() const{
			return mLeftCount * TransContentMemory::getInstance()->getRecordSize();
		}


		int getTimeout() const {
			return this->timeout;
		}
		void setTimeout(int timeout){
			this->timeout = timeout;
		}

		int receive(int flag, void* buffer, int bufferLen, int timeout, recvSyncCallback rscb);
		int getData(char* buffer, int len);
		int response(char* content, int index);

		void setReceiveCallback(recvSyncCallback rscb){
			this->receiveCallback = rscb;
		}

		recvSyncCallback getReceiveCallback() const{
			return this->receiveCallback;
		}

		void setRemoteStreamStart(int startId){
			this->mStreamValueStart = startId;
		}

		int getStreamBufferSize()const{
			return this->mCapacity;
		}

	private:
		///
		/// send ack data to remote devices
		///
		int ackPacket(int index);

		///
		/// write received data to buffer
		///
		int writeData(char* buffer, int index);

		void callbackRecieveData();

		void dataAvaliabkeSignal();

		static void* callbackRoute(void* p);

		void callbackCore();

	public:	
		/// extract information from packet
		///
		/// get main code from packet
		///
		static int getStreamPacketMainCode(const char* buffer) {
			return buffer[0];
		}

		///
		/// get sub code from packet
		///
		static int getStreamPacketSubCode(const char* buffer){
			return buffer[1];
		}

		///
		/// get localPort from packet
		///
		static int getStreamPacketLocalPort(const char* buffer) {
			return TransContentAssis::INTN2L(*(int*(buffer+2)));
		}

		///
		/// get stream value from packet
		///
		static int getStreamPacketStreamValue(const char* buffer) {
			return TransContentAssis::INTN2L(*(int*(buffer+6)));
		}

		///
		/// get data length from packet
		///
		static int getStreamPacketDataLength(const char* buffer) {
			return TransContentAssis::SHORTN2L(*(short*(buffer+10)));
		}

		///
		/// get header length
		///
		static int getStreamPacketHeaderLength(){
			return 12;
		}

		///
		/// get data part start pointer
		///
		static char* getStreamPacketData(char* buffer){
			return buffer+getStreamPacketHeaderLength();
		}
};
#endif //TRANSFORMERCONTENTPORTSTREAMIN_HPP
